package avicit.bdp.dgs.qa.service.engine.sql;

import avicit.bdp.common.service.dto.DataSourceDTO;
import avicit.bdp.common.service.service.DataSourceService;
import avicit.bdp.common.utils.ConfigUtils;
import avicit.bdp.common.utils.DESUtils;
import avicit.bdp.common.utils.database.DBUtils;
import avicit.bdp.core.util.json.JSONUtils;
import avicit.bdp.dds.dispatch.enums.TaskType;
import avicit.bdp.dds.dispatch.process.ResourceInfo;
import avicit.bdp.dds.dispatch.task.shell.ShellParameters;
import avicit.bdp.dgs.qa.dto.JobDTO;
import avicit.bdp.dgs.qa.dto.QaDataSourceDTO;
import avicit.bdp.dgs.qa.dto.RuleDTO;
import avicit.bdp.dgs.qa.service.RuleService;
import avicit.bdp.dgs.qa.service.engine.IEngine;
import avicit.platform6.commons.utils.ComUtil;
import com.alibaba.fastjson2.JSONObject;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.*;

/**
 * @金航数码科技有限责任公司
 * @作者：developer
 * @邮箱：developer@avic-digital.com
 * @创建时间： 2023-09-17
 * @类说明：SqlEngine
 * @修改记录：
 * @注意事项：
 * @主要功能：Sql计算引擎类
 */
@Component("sql")
public class SqlEngine implements IEngine {
    private static final Logger logger = LoggerFactory.getLogger(SqlEngine.class);

    public static final String DEFAULT_JOB_RESULT_TABLE = "bdp_dgs_qa_job_result";

    @Autowired
    private RuleService ruleService;

    @Autowired
    private DataSourceService commDsService;

    /**
     * 数据质量计算结果表连接信息
     */
    @Value("${spring.datasource.dbType:mysql}")
    private String dbType;

    @Value("${spring.datasource.url}")
    private String jdbcUrl;

    @Value("${spring.datasource.driver-class-name}")
    private String driverClass;

    @Value("${spring.datasource.username}")
    private String username;

    @Value("${spring.datasource.password}")
    private String password;

    public SqlEngine() {
    }

    @Override
    public Map<String, Object> build(JobDTO job, Map<String, Object> userOptions) {
        // 构建ProcessDefinition
        Map<String, Object> defeMap = new HashMap<>();

        /**
         * 填充顺序
         * 1、填充Shell计算任务Json配置信息
         * 3、填充TaskNode配置信息
         * 4、填充ProcessDefinition中processDefinitionJson（流程定义信息，格式为List<TaskNode>）
         */
        // step1:填充Task Json新
        Map<String, Object> taskMap = fillTaskMap(job);

        // step2:填充ShellParameters配置信息
        ShellParameters shellParameters = fillShellParameters(job, taskMap);

        // step3:填充TaskNode配置信息
        List<Map<String, Object>> taskNodes = new ArrayList<>();
        taskNodes.add(fillTaskNode(job, shellParameters));

        // step4:填充ProcessDefinition中processDefinitionJson
        defeMap.put("globalParams", new ArrayList<>());
        defeMap.put("tasks", taskNodes);

        return defeMap;
    }

    /**
     * 填充localParameters字段
     *
     * @param job 质量作业
     * @return
     */
    public Map<String, Object> fillTaskMap(JobDTO job) {
        // step1:初始化
        Map<String, Object> localParamsMap = new HashMap<>();
        Map<String, Object> settingParamsMap = new HashMap<>();
        Map<String, Object> parameterMap = new HashMap<>();
        localParamsMap.put("settingParams", settingParamsMap);
        localParamsMap.put("parameter", parameterMap);

        // step2:填充质量任务名称
        fillTaskName(localParamsMap, job);

        // step2:填充task
        fillTaskBaseInfo(parameterMap, job);

        // step5:填充outputConfig
        fillTaskOutputConfig(parameterMap);

        // step6:填充reader/transformer/writer
        fillRuleList(parameterMap, job);
        System.out.println(JSONObject.toJSONString(localParamsMap));

        return localParamsMap;
    }

    /**
     * 填充SparkParameters配置信息
     *
     * @param job
     * @param taskMap
     * @return
     */
    private ShellParameters fillShellParameters(JobDTO job, Map<String, Object> taskMap) {
        ShellParameters shellParameters = new ShellParameters();

        // 设置shell执行搅拌
        shellParameters.setRawScript("java -jar " + ConfigUtils.getInstance().getString("spark.jar.path") + " 1 %s %s");

        // 设置Task Json参数
        ResourceInfo res = new ResourceInfo();
        res.setRes(JSONObject.toJSONString(taskMap));
        shellParameters.setResourceList(Arrays.asList(res));

        return shellParameters;
    }

    /**
     * 填充TaskNode信息
     *
     * @param job
     * @param shellParameters
     * @return
     */
    private Map<String, Object> fillTaskNode(JobDTO job, ShellParameters shellParameters) {
        Map<String, Object> taskMap = new HashMap<>();

        taskMap.put("id", ComUtil.getId());
        taskMap.put("name", job.getName());
        taskMap.put("code", "ALGORITHM");
        taskMap.put("nodeId", ComUtil.getId());
        taskMap.put("color", "#248ad8");
        taskMap.put("iconName", "algorithm");
        taskMap.put("type", TaskType.SHELL.name());
        taskMap.put("preTasks", new ArrayList<>());
        taskMap.put("preTasksObj", new HashMap<>());
        taskMap.put("runFlag", "NORMAL");
        taskMap.put("description", "质量检查任务...");
        taskMap.put("dependence", new HashMap<>());
        taskMap.put("maxRetryTimes", 0);
        taskMap.put("retryInterval", 1);
        taskMap.put("timeout", "");
        taskMap.put("taskInstancePriority", "MEDIUM");
        taskMap.put("workerGroup", "default");
        taskMap.put("params", JSONUtils.toJsonString(shellParameters));

        return taskMap;
    }

    /**
     * 填充Spark校验任务名称
     *
     * @param localParamsMap
     * @param job
     */
    private void fillTaskName(Map<String, Object> localParamsMap, JobDTO job) {
        localParamsMap.put("name", job.getName());
    }

    /**
     * 填充Spark校验任务基本信息，包括jobId、jobName、调度方式、校验类型
     *
     * @param parameterMap
     * @param job
     */
    private void fillTaskBaseInfo(Map<String, Object> parameterMap, JobDTO job) {
        Map<String, Object> taskMap = new HashMap<>();

        taskMap.put("id", job.getId());
        taskMap.put("name", job.getName());
        taskMap.put("schedulerType", job.getSchedulerType());
        taskMap.put("measureType", job.getMeasureType());
        taskMap.put("typeName", job.getTypeName());
        taskMap.put("datasourceName", job.getDatasourceName());
        taskMap.put("dbName", job.getDbName());
        taskMap.put("tableName", job.getTableName());
        taskMap.put("remark", job.getRemark());

        parameterMap.put("task", taskMap);
    }

    /**
     * 填充Spark校验任务输出配置信息
     *
     * @param parameterMap
     */
    private void fillTaskOutputConfig(Map<String, Object> parameterMap) {
        Map<String, Object> outputConfigMap = new HashMap<>();

        outputConfigMap.put("dbType", this.dbType == null ? "mysql" : this.dbType);
        outputConfigMap.put("jdbcUrl", this.jdbcUrl);
        outputConfigMap.put("driverClass", this.driverClass);
        outputConfigMap.put("username", this.username);
        outputConfigMap.put("password", this.password);
        outputConfigMap.put("tableName", DEFAULT_JOB_RESULT_TABLE);

        parameterMap.put("outputConfig", outputConfigMap);
    }

    /**
     * 填充Spark校验任务执行规则，包括reader、transformer、writer信息
     *
     * @param job
     * @param parameterMap
     */
    private void fillRuleList(Map<String, Object> parameterMap, JobDTO job) {
        List<Map<String, Object>> rules = new ArrayList<>();

        List<RuleDTO> ruleList = this.ruleService.getRuleListByJobId(job.getId());
        if (CollectionUtils.isNotEmpty(ruleList)) {
            for (RuleDTO rule : ruleList) {
                if (rule != null) {
                    rules.add(fillSingleRule(rule, job.getDatasourceId(), job.getDbName()));
                }
            }
        }

        parameterMap.put("rules", rules);
    }

    /**
     * 获取单个规则信息
     *
     * @param rule
     * @param datasourceId
     * @param dbName
     * @return
     */
    private Map<String, Object> fillSingleRule(RuleDTO rule, String datasourceId, String dbName) {
        Map<String, Object> configMap = new HashMap<>();

        // 设置规则信息
        setRuleInfo(rule, configMap);

        // 设置数据源信息
        setDatasourceList(rule, datasourceId, dbName, configMap);

        return configMap;
    }

    /**
     * 填充规则基本信息
     *
     * @param rule
     * @param configMap
     */
    private void setRuleInfo(RuleDTO rule, Map<String, Object> configMap) {
        configMap.put("id", rule.getId());
        configMap.put("jobId", rule.getJobId());
        configMap.put("name", rule.getName());
        configMap.put("measureType", rule.getMeasureType());
        configMap.put("measureTypeName", rule.getMeasureTypeName());
        configMap.put("templateType", rule.getTemplateType());
        configMap.put("templateTypeName", rule.getTemplateTypeName());
        configMap.put("templateId", rule.getTemplateId());
        configMap.put("templateName", rule.getTemplateName());
        configMap.put("weight", rule.getWeight());
        configMap.put("weightRatio", rule.getWeightRatio());
        configMap.put("statDataType", rule.getStatDataType());
        configMap.put("alarmFormula", rule.getAlarmFormula());
        configMap.put("dirtyDataType", rule.getDirtyDataType());
        configMap.put("abnormalSql", rule.getAbnormalSql());
    }

    /**
     * 填充规则数据源信息
     *
     * @param rule
     * @param datasourceId
     * @param dbName
     * @param configMap
     */
    private void setDatasourceList(RuleDTO rule, String datasourceId, String dbName, Map<String, Object> configMap) {
        List<Map<String, Object>> dataSourceList = new ArrayList<>();
        if (CollectionUtils.isNotEmpty(rule.getDataSourceList())) {
            // 常见RDMS关系型数据库，支持Druid解析SQL
            setRdmsDatasourceList(rule, datasourceId, dbName, configMap);
        } else {
            //IoTDB数据库，不支持Druid解析SQL，规则数据源可能为空
            setIotdbDatasourceList(rule, datasourceId, dbName, configMap);
        }
    }

    /**
     * 填充规则数据源信息(RDMS)
     *
     * @param rule
     * @param datasourceId
     * @param dbName
     * @param configMap
     */
    private void setRdmsDatasourceList(RuleDTO rule, String datasourceId, String dbName, Map<String, Object> configMap) {
        List<Map<String, Object>> dataSourceList = new ArrayList<>();
        if (CollectionUtils.isNotEmpty(rule.getDataSourceList())) {
            for (QaDataSourceDTO ds : rule.getDataSourceList()) {
                if (ds == null) {
                    continue;
                }

                DataSourceDTO dataSource = this.commDsService.queryDataSourceById(ds.getDatasourceId());
                if (dataSource == null) {
                    logger.error("未找到指定数据源,创建或更新质量任务失败,ruleId=" + rule.getId());
                    continue;
                }
                if (StringUtils.isEmpty(dataSource.getConnectionParams())) {
                    logger.error("指定数据源未配置连接参数,创建或更新质量任务失败,ruleId=" + rule.getId());
                    continue;
                }
                JSONObject connObj = JSONObject.parseObject(dataSource.getConnectionParams());

                Map<String, Object> dsMap = new HashMap<>();
                dsMap.put("id", ds.getId());
                dsMap.put("businessId", ds.getDatasourceType());
                dsMap.put("dbType", ds.getDatasourceType());
                dsMap.put("datasourceId", ds.getDatasourceId());
                dsMap.put("datasourceName", ds.getDatasourceName());
                dsMap.put("dbName", ds.getDbName());
                dsMap.put("tableName", ds.getTableName());
                dsMap.put("field", ds.getField());
                dsMap.put("filter", ds.getFilter());
                dsMap.put("driverClass", DBUtils.getBaseDataSource(dataSource).driverClassSelector());
                dsMap.put("jdbcUrl", connObj.getString("jdbcUrl"));
                dsMap.put("username", connObj.getString("user"));
                dsMap.put("password", DESUtils.decrypt(connObj.getString("password")));
                dsMap.put("ip", connObj.getString("ip"));
                dsMap.put("port", connObj.getString("port"));

                dataSourceList.add(dsMap);
            }
            configMap.put("dataSourceList", dataSourceList);
        }
    }

    /**
     * 填充规则数据源信息(IOTDB)
     *
     * @param rule
     * @param datasourceId
     * @param dbName
     * @param configMap
     */
    private void setIotdbDatasourceList(RuleDTO rule, String datasourceId, String dbName, Map<String, Object> configMap) {
        List<Map<String, Object>> dataSourceList = new ArrayList<>();

        DataSourceDTO dataSource = this.commDsService.queryDataSourceById(datasourceId);
        if (dataSource == null) {
            logger.error("未找到指定数据源,创建或更新质量任务失败,ruleId=" + rule.getId());
            return;
        }
        if (StringUtils.isEmpty(dataSource.getConnectionParams())) {
            logger.error("指定数据源未配置连接参数,创建或更新质量任务失败,ruleId=" + rule.getId());
            return;
        }
        JSONObject connObj = JSONObject.parseObject(dataSource.getConnectionParams());

        Map<String, Object> dsMap = new HashMap<>();
        dsMap.put("businessId", rule.getId());
        dsMap.put("dbType", dataSource.getDatasourceType());
        dsMap.put("datasourceId", dataSource.getId());
        dsMap.put("datasourceName", dataSource.getDatasourceName());
        dsMap.put("dbName", dbName);
        dsMap.put("driverClass", DBUtils.getBaseDataSource(dataSource).driverClassSelector());
        dsMap.put("jdbcUrl", connObj.getString("jdbcUrl"));
        dsMap.put("username", connObj.getString("user"));
        dsMap.put("password", DESUtils.decrypt(connObj.getString("password")));
        dsMap.put("ip", connObj.getString("ip"));
        dsMap.put("port", connObj.getString("port"));

        dataSourceList.add(dsMap);
        configMap.put("dataSourceList", dataSourceList);
    }
}
